package se.sics.kompics.tutorial.tutorial4.AC;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import se.kth.ict.id2203.pp2p.PerfectPointToPointLink;
import se.kth.ict.id2203.pp2p.Pp2pDeliver;
import se.kth.ict.id2203.pp2p.Pp2pSend;
import se.sics.kompics.ComponentDefinition;
import se.sics.kompics.Handler;
import se.sics.kompics.Negative;
import se.sics.kompics.Positive;
import se.sics.kompics.address.Address;
import se.sics.kompics.tutorial.tutorial4.AC.communication.ACPort;
import se.sics.kompics.tutorial.tutorial4.AC.communication.AcPropose;
import se.sics.kompics.tutorial.tutorial4.AC.communication.AcReturn;
import se.sics.kompics.tutorial.tutorial4.AC.communication.NAck;
import se.sics.kompics.tutorial.tutorial4.AC.communication.ReadAck;
import se.sics.kompics.tutorial.tutorial4.AC.communication.ReadMessage;
import se.sics.kompics.tutorial.tutorial4.AC.communication.WriteAck;
import se.sics.kompics.tutorial.tutorial4.AC.communication.WriteMessage;

public class ACImpl extends ComponentDefinition {
	
	//private static final Logger logger = LoggerFactory.getLogger(ACImpl.class);
	
	public static final int ABORT_VALUE = -1;
	
	Negative<ACPort> acPort = provides(ACPort.class);
	Positive<PerfectPointToPointLink> pp2p = requires(PerfectPointToPointLink.class);
	
	private static final Logger logger = LoggerFactory.getLogger(ACImpl.class);
	
	private Set<Address> neighbors;
	private Address self;
	private Set<Integer> seenIds;
	private int majority;
	/*private int[] tempValue;
	private int[] val;
	private int[] wAcks;
	private int[] rts;
	private int[] wts;
	private int[] tstamp;*/
	private Map<Integer, Integer> 	tempValue = new HashMap<Integer, Integer>(), 
									val = new HashMap<Integer, Integer>(), 
									wAcks = new HashMap<Integer, Integer>(), 
									rts = new HashMap<Integer, Integer>(), 
									wts = new HashMap<Integer, Integer>(), 
									tstamp = new HashMap<Integer, Integer>();
	//private Set<ReadSetElement>[] readSet;
	private Map<Integer, Set<ReadSetElement>> readSet = new HashMap<Integer, Set<ReadSetElement>>();
	
	
	public ACImpl() {
		subscribe(handleInit, control);
		subscribe(handleAcPropose, acPort);
		subscribe(handleReadMessage, pp2p);
		subscribe(handleNAck, pp2p);
		subscribe(handleReadAck, pp2p);
		subscribe(handleWriteMessage, pp2p);
		subscribe(handleWriteAck, pp2p);
	}
	
	Handler<ACInit> handleInit = new Handler<ACInit>() {
		
		@Override
		public void handle(ACInit event) {
			self = event.getSelf();
			neighbors = event.getNeighbors();
			neighbors.add(self);
			
			seenIds = new HashSet<Integer>();
			majority = neighbors.size()/2 + 1;
		}
	};
	
	Handler<AcPropose> handleAcPropose = new Handler<AcPropose>() {
		
		@Override
		public void handle(AcPropose event) {
			logger.info("EC-proposing "+event.getV()+" with id "+event.getId());
			initInstance(event.getId());
			//tstamp[event.getId()] += neighbors.size();
			tstamp.put(event.getId(), tstamp.get(event.getId())+neighbors.size());
			//tempValue[event.getId()] = event.getV();
			tempValue.put(event.getId(), event.getV());
			logger.info("PID"+event.getId()+": Broadcasting a ReadRequest with tstamp "+tstamp.get(event.getId()));
			doBroadcast(new ReadMessage(self, event.getId(), tstamp.get(event.getId())));
		}
	};
	
	Handler<ReadMessage> handleReadMessage = new Handler<ReadMessage>() {
		
		@Override
		public void handle(ReadMessage event) {
			int id = event.getId();
			int ts = event.getTstamp();
			initInstance(id);
			
			//if (rts[id] > ts || wts[id] > ts) {
			if(rts.get(id) > ts || wts.get(id) > ts){
				logger.info("PID"+id+": Read request received, responding with NACK");
				trigger(new Pp2pSend(event.getSource(), new NAck(self, id)), pp2p);
			} else {
				// rts[id] = id;
				rts.put(id, ts);
				logger.info("PID"+id+": Read request received, responding with ("+wts.get(id)+", "+val.get(id)+")");
				trigger(new Pp2pSend(event.getSource(),
						new ReadAck(self, id, wts.get(id), val.get(id), ts)), pp2p);
			}
			
		}
	};
	
	Handler<NAck> handleNAck = new Handler<NAck>() {
		
		@Override
		public void handle(NAck event) {
			//readSet[event.getId()] = new HashSet<ReadSetElement>();
			readSet.put(event.getId(), new HashSet<ReadSetElement>());
			//wAcks[event.getId()] = 0;
			wAcks.put(event.getId(), 0);
			logger.info("PID"+event.getId()+": NACK received, aborting the epoch");
			trigger(new AcReturn(event.getId(), ABORT_VALUE), acPort);
		}
	};
	
	Handler<ReadAck> handleReadAck = new Handler<ReadAck>() {
		
		@Override
		public void handle(ReadAck event) {
			int id = event.getId();
			//if (event.getTs() == tstamp[id]) {
			if (event.getTs() == tstamp.get(id)) {
				logger.info("PID"+id+": Received READ response : ("+event.getTs()+", "+event.getVal()+")");
				readSet.get(id).add(new ReadSetElement(event.getTs(), event.getVal()));
				if (readSet.get(id).size() >= majority) {
					ReadSetElement max = highest(readSet.get(id));
					if (max.getV() != ABORT_VALUE) {
						//tempValue[id] = max.getV();
						tempValue.put(id, max.getV());
					}
					logger.info("PID"+id+": Majority of Read collected, bcasting Write message ("+tstamp.get(id)+", "+tempValue.get(id)+")");
					doBroadcast(new WriteMessage(self, id, tstamp.get(id), tempValue.get(id)));
					readSet.put(id, new HashSet<ReadSetElement>());
				}
			}
		}
	};
	
	Handler<WriteMessage> handleWriteMessage = new Handler<WriteMessage>() {
		
		@Override
		public void handle(WriteMessage event) {
			int id = event.getId();
			initInstance(id);
			//if (rts[id] > event.getTstamp() || wts[id] > event.getTstamp()) {
			if (rts.get(id) > event.getTstamp() || wts.get(id) > event.getTstamp()) {
				logger.info("PID"+id+": Write message received ("+event.getTstamp()+", "+event.getTempValue()+"), responding with NACK");
				trigger(new Pp2pSend(event.getSource(), new NAck(self, id)), pp2p);
			} else {
				//val[id] = event.getTempValue();
				val.put(id, event.getTempValue());
				//wts[id] = event.getTstamp();
				wts.put(id, event.getTstamp());
				logger.info("PID"+id+": Write message received ("+event.getTstamp()+", "+event.getTempValue()+"), responding with ACK");
				trigger(new Pp2pSend(event.getSource(), new WriteAck(self, id, event.getTstamp())), pp2p);
			}
		}
	};
	
	Handler<WriteAck> handleWriteAck = new Handler<WriteAck>() {
		
		@Override
		public void handle(WriteAck event) {
			int id = event.getId();
			//if (event.getTs() == tstamp[id]) {
			if (event.getTs() == tstamp.get(id)) {
				//wAcks[id]++;
				wAcks.put(id, wAcks.get(id)+1);
				//if (wAcks[id] >= majority) {
				if (wAcks.get(id) >= majority) {
					//wAcks[id] = 0;
					wAcks.put(id, 0);
					logger.info("PID"+id+": Majority of wACKS, EC-deciding for value "+tempValue.get(id));
					trigger(new AcReturn(id, tempValue.get(id)), acPort);
				}
			}
		}
	};
	
	public void initInstance(int id) {
		if (!seenIds.contains(id)) {
			/*tempValue[id] = ABORT_VALUE;
			val[id] = ABORT_VALUE;
			wAcks[id] = 0;
			rts[id] = 0;
			wts[id] = 0;
			tstamp[id] = self.getPort();*/
			val.put(id, ABORT_VALUE);
			wAcks.put(id, 0);
			rts.put(id, 0);
			wts.put(id, 0);
			tstamp.put(id, self.getPort());
			//readSet[id] = new HashSet<ReadSetElement>();
			readSet.put(id, new HashSet<ReadSetElement>());
			seenIds.add(id);
		}
	}
	
	private void doBroadcast(Pp2pDeliver message) {
		for(Address n : neighbors) {
			trigger(new Pp2pSend(n, message), pp2p);
		}
	}
	
	private ReadSetElement highest(Set<ReadSetElement> set) {
		ReadSetElement max = set.iterator().next();
		for(ReadSetElement element : set) {
			if (element.greaterThan(max)) {
				max = element;
			}
		}
		return max;
	}
	
	private class ReadSetElement {
		
		private int ts;
		private int v;
		
		public ReadSetElement(int ts, int v) {
			this.ts = ts;
			this.v = v;
		}
		
		public int getTs() {
			return ts;
		}
		
		public int getV() {
			return v;
		}
		
		public boolean greaterThan(ReadSetElement other) {
			if (ts != other.getTs()) {
				return (ts > other.getTs());
			} else {
				return (v > other.getV());
			}
		}
	}
}
